package queue

import (
	"context"
	"fmt"
	"github.com/gogf/gf/v2/container/gmap"
	"github.com/gogf/gf/v2/container/gqueue"
	"github.com/gogf/gf/v2/container/gtype"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/gogf/gf/v2/util/gutil"
	"math/rand"
	"time"
)

var (
	delayQueueHandledFlags = gmap.NewStrIntMap(true)
	wrrQueues              = map[string]*gqueue.Queue{}
)

const delayQueueluaScript = `
local jobs= redis.call('ZRANGEBYSCORE',KEYS[1],ARGV[1],ARGV[2])
if #jobs >0 then
	redis.call('LPUSH',KEYS[2], unpack(jobs))
	redis.call('ZREMRANGEBYSCORE',KEYS[1],ARGV[1],ARGV[2])
end
`

func getQueueName(name string, priority QueuePriority) string {
	return fmt.Sprintf("easy-queue-%s-%s", name, priority)
}
func GetQueueLength(name string, priority ...QueuePriority) int {
	p := QueuePriorityDefault
	if len(priority) > 0 {
		p = priority[0]
	}
	queueName := getQueueName(name, p)
	length, err := easyRedis.Ctx(context.Background()).LLen(queueName)
	if err != nil {
		glog.Error(nil, fmt.Sprintf("failed to get queue length: %s", err.Error()))
	}
	return length.Int()
}
func getSortedQueues(queueName string, strictPriority bool, priorities *gmap.ListMap) []interface{} {
	if strictPriority {
		return append(priorities.Keys(), queueName)
	}

	// 当队列不存在时，进行初始化
	if _, ok := wrrQueues[queueName]; !ok {
		values := priorities.Values()
		// 计算最大权重
		weight := 0
		for _, value := range values {
			weight = weight + gconv.Int(value)
		}
		// 初始化队列（gqueue也支持固定队列大小，固定队列大小时队列效率和标准库的channel无异。）
		wrrQueues[queueName] = gqueue.New(weight)
	}

	var args []interface{}

	// 当队列长度为0时，将队列名称使用洗牌算法打散，然后重新入队
	if wrrQueues[queueName].Len() == 0 {
		var queueNames []interface{}
		priorities.Iterator(func(k, v interface{}) bool {
			for i := 0; i < gconv.Int(v); i++ {
				queueNames = append(queueNames, k)
			}
			return true
		})

		r := rand.New(rand.NewSource(time.Now().UnixNano()))
		r.Shuffle(len(queueNames),
			func(i, j int) {
				queueNames[i], queueNames[j] = queueNames[j], queueNames[i]
			},
		)

		for _, name := range queueNames {
			wrrQueues[queueName].Push(name)
		}
	}

	// 弹出一个队列名称
	popName := wrrQueues[queueName].Pop()

	args = append(args, popName)
	priorities.Iterator(func(k, v interface{}) bool {
		if gconv.String(k) != popName {
			args = append(args, k)
		}
		return true
	})
	return append(args, queueName)
}

func Handle(queueName string, canConsume func(runningJobCount int64) bool, doJob func(job *QueueJob), opts ...HandleOption) {
	go func() {
		// 初始化队列优先级权重
		priorities := gmap.NewListMap()
		priorities.Set(getQueueName(queueName, QueuePriorityHigh), 6)
		priorities.Set(getQueueName(queueName, QueuePriorityDefault), 3)
		priorities.Set(getQueueName(queueName, QueuePriorityLow), 1)

		strictPriority := false
		for _, opt := range opts {
			switch opt := opt.(type) {
			case handleOptionStrictPriority:
				strictPriority = bool(opt)
			}
		}
		runningJobCount := gtype.NewInt64()

		for {
			if !canConsume(runningJobCount.Val()) {
				time.Sleep(time.Second)
				continue
			}

			sortedQueues := getSortedQueues(queueName, strictPriority, priorities)
			// 5 表示 block 5秒钟
			sortedQueues = append(sortedQueues, 5)

			v, err := easyRedis.Ctx().BRPop(sortedQueues...)

			if err == nil {
				if !v.IsEmpty() {
					runningJobCount.Add(1)
					go func() {
						// gutil.Try 确保即使业务代码异常后，也不能影响计数操作
						err = gutil.Try(context.Background(), func(ctx context.Context) {
							data := v.Slice()[1]
							var item *QueueJob
							gconv.Scan(data, &item)
							item.HandledAt = gtime.Now().TimestampMilli()
							doJob(item)
						})
						if err != nil {
							glog.Error(nil, "failed to do job", err)
						}
						runningJobCount.Add(-1)
					}()
				}
			} else {
				time.Sleep(time.Second)
			}
		}
	}()
}
func Dispatch(queueName string, job interface{}, opts ...DispatchOption) error {
	priority := QueuePriorityDefault

	processIn := 0
	useRawQueueName := false
	for _, opt := range opts {
		switch opt := opt.(type) {
		case dispatchOptionPriority:
			priority = QueuePriority(opt)
		case dispatchOptionProcessIn:
			processIn = int(opt)
		case dispatchOptionUseRawQueueName:
			useRawQueueName = bool(opt)
		}
	}

	queue := getQueueName(queueName, priority)
	if useRawQueueName {
		queue = queueName
	}

	message := newQueueJob(job, priority)

	if processIn <= 0 {
		_, err := easyRedis.Ctx().LPush(queue, message)
		if err != nil {
			return err
		}
	} else {
		delayQueue := fmt.Sprintf("easy-queue-%s-delay", queueName)
		score := message.DispatchedAt + int64(processIn)*1000
		_, err := easyRedis.Ctx().ZAdd(delayQueue, score, message)
		if err != nil {
			return err
		}

		defer func() {
			if !delayQueueHandledFlags.Contains(queueName) {
				delayQueueHandledFlags.Set(queueName, 1)
				go func() {
					for {
						nowMilliseconds := gtime.Now().TimestampMilli()
						easyRedis.Ctx().Eval(delayQueueluaScript, 2, delayQueue, queue, 0, nowMilliseconds)
						time.Sleep(time.Second)
					}
				}()
			}
		}()
	}
	return nil
}
